package net.bwie.realtime.jtp.dws.douyin.log.job;
import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.FanRelation1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.LiveStatMetric1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.PayDetail1;
import net.bwie.realtime.jtp.dws.douyin.log.bean.LiveUnifiedData1;
import net.bwie.realtime.jtp.dws.douyin.log.functions.LiveStatWindowFunction1;
import net.bwie.realtime.jtp.utils.DorisUtil;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
public class DouYinLiveStat1 {
    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink环境（与原有类完全一致）
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);          // 测试并行度1
        env.enableCheckpointing(3000L); // 3秒Checkpoint

        // 2. 读取3张表的Kafka数据源（对应3个主题）
        DataStream<String> payKafkaStream = KafkaUtil.consumerKafka(env, "ods_trade_pay_detail"); // 支付表主题
        DataStream<String> fanKafkaStream = KafkaUtil.consumerKafka(env, "ods_user_fan_relation"); // 粉丝关系表主题
        DataStream<String> logKafkaStream = KafkaUtil.consumerKafka(env, "ods_live_event_log");    // 直播日志表主题

        // 3. 数据处理：解析→封装载体→合并→窗口聚合→JSON格式化
        DataStream<String> resultStream = handle(payKafkaStream, fanKafkaStream, logKafkaStream);

        // 4. 打印结果（测试用，观察指标格式）
        resultStream.print("直播统计指标→");

        // 5. 数据输出到Doris（与原有类一致，数据库/表名适配直播指标）
        DorisUtil.saveToDoris(
                resultStream,
                "douyin_realtime_report",  // 数据库名（与原有一致）
                "dws_live_stat_detail"    // 直播统计结果表名
        );

        // 6. 触发作业执行（必须调用，否则无算子执行）
        env.execute("DouYinLiveStat1");
    }

    /**
     * 核心处理方法（参照原有handle结构，步骤化处理3张表数据）
     */
    private static DataStream<String> handle(
            DataStream<String> payStream,
            DataStream<String> fanStream,
            DataStream<String> logStream) {

        // ------------------------------ 步骤1：解析支付表→封装为LiveUnifiedData（flag=1） ------------------------------
        SingleOutputStreamOperator<LiveUnifiedData1> payUnifiedStream = payStream
                // 解析JSON为PayDetail6，支付金额转为double（替代BigDecimal）
                .map(jsonStr -> {
                    PayDetail1 pay = JSON.parseObject(jsonStr, PayDetail1.class);
                    // 处理金额：JSON中pay_amount（DECIMAL）→ double
                    String amountStr = JSON.parseObject(jsonStr).getString("pay_amount");
                    pay.setPayAmount(amountStr != null ? Double.parseDouble(amountStr) : 0.0);
                    return pay;
                })
                .filter(pay -> pay != null) // 过滤空数据
                .filter(pay -> "success".equals(pay.getPayStatus())) // 仅保留成功支付数据
                // 封装为LiveUnifiedData：flag=1（支付数据）
                .map(pay -> {
                    LiveUnifiedData1 unifiedData = new LiveUnifiedData1();
                    unifiedData.setDataTypeFlag(1);
                    unifiedData.setPayData(pay);
                    unifiedData.setFanData(null);
                    unifiedData.setLogData(null);
                    return unifiedData;
                });

        // ------------------------------ 步骤2：解析粉丝关系表→封装为LiveUnifiedData（flag=2） ------------------------------
        SingleOutputStreamOperator<LiveUnifiedData1> fanUnifiedStream = fanStream
                // 解析JSON为FanRelation6（直接映射表字段）
                .map(jsonStr -> JSON.parseObject(jsonStr, FanRelation1.class))
                .filter(fan -> fan != null) // 过滤空数据
                .filter(fan -> fan.getChangeTime() != null) // 过滤无关系变化时间的数据
                // 封装为LiveUnifiedData：flag=2（粉丝关系数据）
                .map(fan -> {
                    LiveUnifiedData1 unifiedData = new LiveUnifiedData1();
                    unifiedData.setDataTypeFlag(2);
                    unifiedData.setFanData(fan);
                    unifiedData.setPayData(null);
                    unifiedData.setLogData(null);
                    return unifiedData;
                });

        // ------------------------------ 步骤3：解析直播日志表→封装为LiveUnifiedData（flag=3） ------------------------------
        SingleOutputStreamOperator<LiveUnifiedData1> logUnifiedStream = logStream
                // 解析JSON为EventLog6（与原有DouYinOrderSourceRatio3一致）
                .map(jsonStr -> JSON.parseObject(jsonStr, EventLog6.class))
                .filter(log -> log != null) // 过滤空数据
                // 仅保留直播相关事件（进入/退出直播间、在线人数统计）
                .filter(log -> "进入直播间".equals(log.getEventType())
                        || "退出直播间".equals(log.getEventType())
                        || "在线人数统计".equals(log.getEventType()))
                // 封装为LiveUnifiedData：flag=3（直播日志数据）
                .map(log -> {
                    LiveUnifiedData1 unifiedData = new LiveUnifiedData1();
                    unifiedData.setDataTypeFlag(3);
                    unifiedData.setLogData(log);
                    unifiedData.setPayData(null);
                    unifiedData.setFanData(null);
                    return unifiedData;
                });

        // ------------------------------ 步骤4：合并3张表的UnifiedData流 ------------------------------
        DataStream<LiveUnifiedData1> mergedStream = payUnifiedStream
                .union(fanUnifiedStream)
                .union(logUnifiedStream);

        // ------------------------------ 步骤5：按主播ID分组+滑动窗口（与原有策略一致） ------------------------------
        WindowedStream<LiveUnifiedData1, Long, TimeWindow> windowStream = mergedStream
                // 分组键：主播ID（与原有一致，按主播维度统计直播指标）
                .keyBy(unifiedData -> {
                    // 从不同数据类型中提取主播ID
                    if (unifiedData.getDataTypeFlag() == 1 && unifiedData.getPayData() != null) {
                        return unifiedData.getPayData().getAnchorId();
                    } else if (unifiedData.getDataTypeFlag() == 2 && unifiedData.getFanData() != null) {
                        return unifiedData.getFanData().getAnchorId();
                    } else if (unifiedData.getDataTypeFlag() == 3 && unifiedData.getLogData() != null) {
                        return unifiedData.getLogData().getAnchorId();
                    }
                    return 0L; // 异常数据归为0组
                })
                // 滑动窗口：5秒窗口+1秒滑动（测试用，生产改为5分钟窗口+1分钟滑动）
                .window(SlidingProcessingTimeWindows.of(
                        Time.seconds(5),
                        Time.seconds(1)
                ));

        // ------------------------------ 步骤6：应用窗口函数，计算直播指标 ------------------------------
        SingleOutputStreamOperator<LiveStatMetric1> metricStream = windowStream
                .apply(new LiveStatWindowFunction1()); // 自定义直播指标窗口函数

        // ------------------------------ 步骤7：格式化指标为JSON（适配Doris写入，处理空值） ------------------------------
        SingleOutputStreamOperator<String> resultStream = metricStream
                .map(metric -> String.format(
                        "{\"window_start_time\":\"%s\",\"window_end_time\":\"%s\",\"cur_date\":\"%s\",\"anchor_id\":\"%d\",\"live_id\":\"%s\",\"total_deal_amount\":\"%.2f\",\"total_deal_goods_count\":\"%d\",\"deal_user_id_count\":\"%d\",\"deal_conversion_rate\":\"%.2f\",\"per_thousand_view_amount\":\"%.2f\",\"deal_fan_ratio\":\"%.2f\",\"avg_online_user_count\":\"%.2f\",\"total_view_user_count\":\"%d\",\"new_fan_club_count\":\"%d\",\"new_fan_count\":\"%d\",\"avg_watch_duration\":\"%.2f\"}",
                        // 空值处理（与原有类一致，字符串用空串，数值用0）
                        metric.getWindowStartTime() != null ? metric.getWindowStartTime() : "",
                        metric.getWindowEndTime() != null ? metric.getWindowEndTime() : "",
                        metric.getCurDate() != null ? metric.getCurDate() : "",
                        metric.getAnchorId() != null ? metric.getAnchorId() : 0,
                        metric.getLiveId() != null ? metric.getLiveId() : "",
                        metric.getTotalDealAmount() >= 0 ? metric.getTotalDealAmount() : 0.00,
                        metric.getTotalDealGoodsCount() ,
                        metric.getDealUserIdCount(),
                        metric.getDealConversionRate() >= 0 ? metric.getDealConversionRate() : 0.00,
                        metric.getPerThousandViewAmount() >= 0 ? metric.getPerThousandViewAmount() : 0.00,
                        metric.getDealFanRatio() >= 0 ? metric.getDealFanRatio() : 0.00,
                        metric.getAvgOnlineUserCount(),
                        metric.getTotalViewUserCount(),
                        metric.getNewFanClubCount(),
                        metric.getNewFanCount(),
                        metric.getAvgWatchDuration() >= 0 ? metric.getAvgWatchDuration() : 0.00
                ));

        return resultStream;
    }
}
